package TestSparkStreaming;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.streaming.Duration;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;

public class TestReduceByWindow {
    public static void main(String[] args) throws InterruptedException {
        SparkConf sparkConf = new SparkConf().setAppName("TestReduceByWindow").setMaster("local[2]");

        JavaSparkContext javaSparkContext = new JavaSparkContext(sparkConf);

        JavaStreamingContext javaStreamingContext = new JavaStreamingContext(javaSparkContext, new Duration(1000));

        JavaReceiverInputDStream<String> javaReceiverInputDStream = javaStreamingContext.socketTextStream("localhost", 9999);

        JavaDStream<String> reduceByWindow = javaReceiverInputDStream.reduceByWindow(new Function2<String, String, String>() {
            @Override
            public String call(String s, String s2) throws Exception {
                return s + "+" + s2;
            }
        }, new Duration(1000 * 6), new Duration(1000 * 3));

        reduceByWindow.print();

        javaStreamingContext.start();
        javaStreamingContext.awaitTermination();
    }

}
